package com.quanyan.club.mq.consumer;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.api.APIResponse;
import com.quanyan.club.mq.constant.MqConstant;
import com.quanyan.club.service.ClubActivityService;
import com.quanyan.common.constants.Constants;
import com.quanyan.common.utils.ObjectUtil;
import com.quanyan.mq.enums.MqConsumerStatus;
import com.quanyan.mq.mq.consumer.AbstractMqConsumer;
import com.quanyan.mq.mq.consumer.MqTopic;
import com.quanyan.orderpay.response.mq.OrderNotifyMsg;
import com.quanyan.orderpay.service.OrderPayService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;

import javax.annotation.PostConstruct;

/**
 * @Author Hugo.Wwg
 * @Since 2016-06-25
 */
@Component
public class ActivityOrderCloseConsumer extends AbstractMqConsumer {

    private static final Logger log = LoggerFactory.getLogger(ActivityOrderCloseConsumer.class);

    @Autowired
    private ClubActivityService clubActivityService;

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    @Override
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        MessageExt messageExt = messages.get(0);
        String msgId = messageExt.getKeys();
        byte[] bodyByte = messageExt.getBody();

        try {
            log.warn("mq收到支付系统订单关闭消息");
            OrderNotifyMsg orderNotifyMsg = (OrderNotifyMsg) ObjectUtil.toObject(bodyByte);
            APIResponse apiResponse = clubActivityService.
                    handleActivityOrderCloseCallBack(orderNotifyMsg);
            if (!apiResponse.isRet()) {
                //todo 消费异常,记录msgId 到 db 中

                return MqConsumerStatus.CONSUMER_SUCCESS;
            }
        } catch (Exception e) {
            //  todo 消费异常,记录msgId 到 db 中
            log.error("消费异常，msgId【{}】，topic【{}】,tag【{}】,errorMsg【{}】", msgId,
                    OrderPayService.TOPIC_ORDER_STATUS_NOTIFY,
                    Constants.CLUB_SYSTEM_CODE, e.getMessage());

            return MqConsumerStatus.CONSUMER_SUCCESS;
        }

        log.warn(Thread.currentThread().getName() + " receive messages: " + messages.toString());

        return MqConsumerStatus.CONSUMER_SUCCESS;
    }

//    @PostConstruct
//    void run() {
//        log.warn("nameServerAddress : " + nameServerAddress);
//        super.start();
//    }

    @Override
    protected String getConsumerGroup() {
        return MqConstant.CLUB_ACTIVITY_ORDER_CLOSE_CALLBACK_GROUP;
    }

    @Override
    protected String getNameServerAddress() {
        return this.nameServerAddress;
    }

    @Override
    protected String getInstanceName() {
        return MqConstant.CLUB_ACTIVITY_ORDER_CLOSE_CALLBACK_INSTANCE_NAME;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(
                OrderPayService.TOPIC_ORDER_CLOSE_STATUS_NOTIFY, Constants.CLUB_SYSTEM_CODE);
        return mqTopics;
    }
}
